#!usr/bin/env python  
# -*- coding:utf-8 _*-
""" 
@author:robot
@file: tpfv1_dataflow.py 
@version:
@time: 2022/01/11 
"""
from TapException.TM import *
from TapUtil.decorator.some_useful import timeout
from TapUtil.logfunc.logger import logger
from .tpf_dataflow_base import TPFDataFlowBase


class V1TPFDataFlow(TPFDataFlowBase):
    dataflow_type = None
    dataflow_url = None
    DATAFLOW_FUNC = {
        'data clone': ...,
        'data sync': ...,
    }

    def _create_data_transmission(self, data):
        record = {}
        create_response = None
        dataflow_type = self.dataflow_type
        dataflow_config = data.get('dataflow_config')
        dataflow_func = self.DATAFLOW_FUNC.get(dataflow_type)
        if dataflow_func is None:
            raise DataFlowBuildError
        DAG = dataflow_func(dataflow_config)
        create_response = self.TMF.create_dataflow(DAG)
        # 更新
        create_response = self.TMF.patch_dataflow(create_response)
        record['id'] = create_response.get('id')
        record['DAG'] = DAG
        record['create_response'] = create_response
        return record

    def create_data_copy_development(self, data):
        self.dataflow_type = data.get('dataflow_type')
        dataflow_name = data.get('dataflow_config').get('config').get('name')
        repeat = self.TMF.query_dataflow_by_name(dataflow_name)
        # query_result = self.dataflow_query_mongodb({'name': dataflow_name})
        if repeat:
            logger.info("同名DataFLow已存在")
            dataflow_status = self.TMF.query_dataflow_by_name(dataflow_name).get('status')
            if dataflow_status == 'running':
                self.TMF.stopping_and_wait_to_paused(dataflow_name)
            try:
                self.TMF.delete_dataflow_by_name(dataflow_name)
                self.dataflow_delete_mongodb({'name': dataflow_name})
            except Exception as e:
                assert False
        logger.info(f"{dataflow_name}创建任务，并将记录持久化")
        record = self._create_data_transmission(data)
        self.dataflow_insert_mongodb(
            {'name': dataflow_name, **record}
        )
        return record

    def _delete_dataflow(self, name: str) -> None:
        status = self.TMF.query_dataflow_some_by_name(
            name,
            {'status': True},
            get='status')
        logger.info("DELETE DATAFLOW START")
        if status in ['stop', 'error', 'draft']:
            self.TMF.delete_dataflow_by_name(name)
        else:
            self.TMF.stopping_and_wait_to_paused(name)
            self.TMF.delete_dataflow_by_name(name)
        logger.info("DELETE DATAFLOW END")

    def delete_dataflow(self, name: str) -> None:
        """
        delete dataflow
        :param name:
        :return:
        """
        try:
            self._delete_dataflow(name)
        except DataFlowNotFound as e:
            logger.info(e)
        except DataFlowRemoveFail as e:
            logger.fatal(e)
            assert False
        self.dataflow_delete_mongodb({'name': name})

    def start_dataflow_wait_running(self, name):
        self.TMF.start_and_wait_to_running(name)

    def start_dataflow_wait_paused(self, name):
        self.TMF.start_and_wait_to_paused(name)

    def stop_dataflow_wait_paused(self, name):
        self.TMF.stopping_and_wait_to_paused(name)

    def generate_dataflows_url(self, name):
        _id = self.TMF.query_dataflow_some_by_name(name)
        if 'data clone' in self.dataflow_type:
            self.dataflow_url = f'{self.serve}/#/job?id={_id}&isMonitin=true&mapping=cluster-clone'
        elif 'data sync' in self.dataflow_type:
            self.dataflow_url = f'{self.serve}/#/job?id={_id}&isMonitin=true&mapping=custom'
        else:
            raise 'dataflow_type error'

    @timeout('GET STATS', 20)
    def _get_dataflow_stats(self, name):
        stats = self.TMF.query_dataflow_some_by_name(
            name,
            {'stats': True},
            'stats'
        )
        if stats:
            if stats.get('input'):
                self.dataflow_stats = stats
                return 'GET STATS'

    def get_dataflow_stats(self, name):
        self._get_dataflow_stats(name)
        return self.dataflow_stats

    def get_dataflow_stats_insert(self, name):
        return self.get_dataflow_stats(name).get('insert').get('rows')

    def get_dataflow_stats_update(self, name):
        return self.get_dataflow_stats(name).get('update').get('rows')

    def get_dataflow_stats_delete(self, name):
        return self.get_dataflow_stats(name).get('delete').get('rows')

    def get_dataflow_stats_insert_datasize(self, name):
        return self.get_dataflow_stats(name).get('insert').get('dataSize')

    def get_dataflow_stats_update_datasize(self, name):
        return self.get_dataflow_stats(name).get('update').get('dataSize')

    def get_dataflow_stats_delete_datasize(self, name):
        return self.get_dataflow_stats(name).get('delete').get('dataSize')
